package cn.com.rocketmq.transaction;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;

/**
 * 本地事务实现类
 * @author xuhaixiang
 * @date 2023-09-11
 */
public class TransactionListenerImpl implements TransactionListener {

    /**
     * 日志对象
     */
    private static  final Logger logger = LoggerFactory.getLogger(TransactionListenerImpl.class);

    /**
     * 本地事务执行
     * @param message
     * @param o
     * @return
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        String tags = message.getTags();
        logger.info("{}本地事务执行", tags);
        if ("tagA".equals(tags)) {
            //tagA允许发送
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        if ("tagB".equals(tags)) {
            //tagB消息回滚
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        //其他消息无状态，无状态消息会进行本地事务回查
        return LocalTransactionState.UNKNOW;
    }

    /**
     * 本地事务回查
     * @param messageExt
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        String tags = messageExt.getTags();
        logger.info("{}本地事务回查", tags);
        if ("tagC".equals(tags)) {
            //tagC本地事务回查允许发送
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.UNKNOW;
    }
}
